package com.starnet.yarnmonitor.yarn.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.starnet.yarnmonitor.yarn.conf.HDFSConf;
import com.starnet.yarnmonitor.yarn.handler.ApiException;
import com.starnet.yarnmonitor.yarn.utils.CompressUtils;
import com.starnet.yarnmonitor.yarn.vo.AppVO;
import com.starnet.yarnmonitor.yarn.vo.MetricsVO;
import com.starnet.yarnmonitor.yarn.vo.NodeVO;
import com.starnet.yarnmonitor.yarn.service.ClusterService;
import com.starnet.yarnmonitor.yarn.utils.HttpUtils;
import com.starnet.yarnmonitor.yarn.vo.QueueVO;
import lombok.extern.log4j.Log4j;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.regex.Pattern;
@Log4j
@Service
public class ClusterServiceImpl implements ClusterService {
    @Autowired
    HttpUtils httpUtils;
    @Autowired
    YarnClient myYARN;
    @Autowired
    FileSystem myHDFS;
    @Autowired
    HDFSConf hdfsConf;

    @Override
    public List<NodeVO> getAllNodes(Integer cur, Integer size) throws IOException {
        String result = null;
            result = httpUtils.get(HttpUtils.NODES_URL);
        JSONObject jsonObject = JSON.parseObject(result);
        JSONArray jsonArray = jsonObject.getJSONObject("nodes").getJSONArray("node");
        List<NodeVO> resultNodeVOS = new ArrayList<>();
        for (int i = 0; i < jsonArray.size(); i++) {
            //当前node的json对象
            JSONObject jsonNode = jsonArray.getJSONObject(i);
            //node里面的resourceUtilization
            JSONObject resourceUtilization = jsonNode.getJSONObject("resourceUtilization");//
            //当前node的java对象
            NodeVO nodeVO = JSON.parseObject(jsonArray.getString(i), NodeVO.class);
            nodeVO.setNodePhysicalMemoryMB(resourceUtilization.getInteger("nodePhysicalMemoryMB"));
            nodeVO.setNodeVirtualMemoryMB(resourceUtilization.getInteger("nodeVirtualMemoryMB"));
            nodeVO.setNodeCPUUsage(resourceUtilization.getDouble("nodeCPUUsage"));
            nodeVO.setAggregatedContainersPhysicalMemoryMB(resourceUtilization.getInteger("aggregatedContainersPhysicalMemoryMB"));
            nodeVO.setAggregatedContainersVirtualMemoryMB(resourceUtilization.getInteger("aggregatedContainersVirtualMemoryMB"));
            nodeVO.setContainersCPUUsage(resourceUtilization.getDouble("containersCPUUsage"));
            resultNodeVOS.add(nodeVO);
        }
        //分页后的返回结果

        return resultNodeVOS;

    }

    @Override
    public MetricsVO getClusterInfo() throws IOException {
        String httpRes = null;
            httpRes = httpUtils.get(HttpUtils.METRICS_URL);
        JSONObject jsonRes = JSON.parseObject(httpRes);
        //获得里面一层clusterMetrics,然后解析成Metrics的java对象
        String clusterMetricsStr = jsonRes.getString("clusterMetrics");
        MetricsVO metrics = JSON.parseObject(clusterMetricsStr, MetricsVO.class);
        return metrics;

    }

    @Override
    public List<QueueVO> getAllQueues(Integer cur, Integer size) throws IOException, YarnException {
        QueueVO queueVO;
        //返回的类resList
        List<QueueVO> resList = new ArrayList<>();
        List<QueueInfo> allQueues = null;
            allQueues = myYARN.getAllQueues();
        for (QueueInfo queueInfo : allQueues) {
            queueVO = new QueueVO();
            //queueStatistics内容numAppsSubmitted: 1 numAppsRunning: 0 numAppsPending: 0 numAppsCompleted: 1 numAppsKilled: 0 numAppsFailed: 0
            // numActiveUsers: 0 availableMemoryMB: 24576 allocatedMemoryMB: 0 pendingMemoryMB: 0 reservedMemoryMB: 0
            //availableVCores: 24 allocated Cores: 0 pendingVCores: 0 reservedVCores: 0 allocatedContainers: 0 pendingContainers: 0 reservedContainers: 0
            QueueStatistics queueStatistics = queueInfo.getQueueStatistics();
            //从api调用结果拷贝值到vo
            BeanUtils.copyProperties(queueInfo, queueVO);
            BeanUtils.copyProperties(queueStatistics, queueVO);
            queueVO.setState(queueInfo.getQueueState().toString());
            resList.add(queueVO);
        }

        return resList;
    }

    @Override
    public List<AppVO> getAllApps(Integer cur, Integer size) throws IOException, YarnException {
        List<ApplicationReport> applications = null;
        //返回的结果
        List<AppVO> resList = new ArrayList<>();
        applications = myYARN.getApplications();
        if (applications.size() == 0) {
            return null;
        }
        AppVO appVo;
        for (ApplicationReport applicationReport : applications) {
            appVo = new AppVO();
            BeanUtils.copyProperties(applicationReport, appVo);
            appVo.setYarn_application_state(applicationReport.getYarnApplicationState().toString());
            appVo.setFinal_application_status(applicationReport.getFinalApplicationStatus().toString());
            appVo.setPriority(applicationReport.getPriority().toString());
            appVo.setApplicationId(applicationReport.getApplicationId().toString());
            resList.add(appVo);
        }



        return resList;

    }



    @Override
    public void getLogsByAppId(HttpServletResponse response, String applicationId) throws ApiException {
        //1.从hdfs下载文件到项目内的临时文件夹
        //获取要下载的文件的路径
        List<String> logFileList = new ArrayList<>();
        //回传后待删除的目录
        List<String> deleteList = new ArrayList<>();
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = null;
        try {
            locatedFileStatusRemoteIterator = myHDFS.listFiles(new Path(hdfsConf.getLogPath()), true);
            while (locatedFileStatusRemoteIterator.hasNext()) {
                LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
                String curPath = next.getPath().toString();
                String pattern = ".*" + applicationId + ".*";
                //正则匹配 如果带有这个appId 就是目标文件 添加到待下载列表
                boolean isMatch = Pattern.matches(pattern, curPath);
                if (isMatch) {
                    logFileList.add(curPath);
                }
            }

            if (logFileList.size() == 0) {
                throw new ApiException("无匹配的applicaitionID");
            }
            //从hdfs上下载到临时文件夹
            File hdfstmpFile = new File("./src/main/resources/tmp/hdfstmp");
            File ziptmpFile = new File("./src/main/resources/tmp/ziptmp");
            //临时文件夹路径
            String hdfstmp = hdfstmpFile.getCanonicalPath();
            String ziptmp = ziptmpFile.getCanonicalPath();
            String uid = String.valueOf(UUID.randomUUID());
            for (String logPath : logFileList) {
                Path src = new Path(logPath);
                //后缀
                String suffix = logPath.substring(logPath.lastIndexOf('.'));
                //临时文件名
                String dstPath = hdfstmp + "/" + applicationId + "_" + uid + suffix;
                deleteList.add(dstPath);
                Path dst = new Path(dstPath);
                myHDFS.copyToLocalFile(src, dst);
            }
            List<String> sourceFilePaths = new ArrayList<>();
            for (String toZipPath : deleteList) {
                sourceFilePaths.add(toZipPath);
            }
            //压缩文件名
            String zipPath = ziptmp + "/" + applicationId + "_" + uid + ".zip";
            deleteList.add(zipPath);
            //压缩过程
            int s = CompressUtils.compress(sourceFilePaths, zipPath, false);

            //回传部分
            response.setContentType("application/octet-stream");
            response.addHeader("Content-Disposition", "filename=" + applicationId + "_" + uid + ".zip");
            //设置文件名，设置字符集是避免文件名中有中文时出现乱码
            InputStream is = null;
            is = new FileInputStream(new File(zipPath));
            OutputStream outputStream = response.getOutputStream();


//            byte[] buffer = new byte[1024];
//            int len =0;
//            while ((len=is.read(buffer))!=1){
//                outputStream.write(buffer);
//            }

            outputStream.write(is.readAllBytes());
            outputStream.close();
            is.close();
            //删除刚刚下载的临时目录内的内容
            for (String deletePath : deleteList) {
                File df = new File(deletePath);
                if (df.exists()) {
                    df.delete();
                }
            }
            //删除crc文件
            deleteCRC();
        } catch (IOException e) {
            log.error("hdfs宕机");
            e.printStackTrace();
            throw new ApiException("hdfs宕机");
        }

    }

    @Override
    public AppVO getAppById(String applicationId) throws IOException {
        applicationId = "application_" + applicationId;
        ApplicationReport applicationReport = null;
        try {
            applicationReport = myYARN.getApplicationReport(ApplicationId.fromString(applicationId));
        } catch (IOException e) {
            log.error(e.getStackTrace());
            throw new ApiException("yarn连接错误，yarn可能宕机了");
        }catch (IllegalArgumentException e){
            log.error(e.getStackTrace());
            throw new ApiException("不存在此id应用");
        } catch (YarnException e) {
            log.error(e.getStackTrace());
            throw new ApiException("yarn连接错误，yarn可能宕机了");
        }
        if (applicationReport == null) {
            throw new ApiException("不存在该应用");
        }
        AppVO appVo = new AppVO();
        BeanUtils.copyProperties(applicationReport, appVo);
        appVo.setYarn_application_state(applicationReport.getYarnApplicationState().toString());
        appVo.setFinal_application_status(applicationReport.getFinalApplicationStatus().toString());
        appVo.setPriority(applicationReport.getPriority().toString());
        appVo.setApplicationId(applicationReport.getApplicationId().toString());
        return appVo;
    }

    public static void deleteCRC() {
        File hdfstmpFile = new File("./src/main/resources/tmp/hdfstmp");
        File[] files = hdfstmpFile.listFiles();
        String crcPattern = ".*.crc";
        for (File f : files) {
            boolean isCRC = Pattern.matches(crcPattern, f.getName());
            if (isCRC) {
                if (f.exists()) {
                    f.delete();
                }
            }
        }
    }
}
